package com.xiongmm.stream;

import com.alibaba.fastjson.JSON;
import com.ucar.flinkcomponent.connectors.flexq.FlexQConstant;
import com.ucar.flinkcomponent.connectors.flexq.FlexQSource;
import com.ucar.flinkcomponent.connectors.opentsdb.OpentsdbSink;
import com.ucar.opentsdb.client.value.request.Point;
import com.xiongmm.stream.domain.InstanceInfoAllVo;
import com.xiongmm.stream.domain.InstanceInfoCmdVo;
import com.xiongmm.stream.sink.SinkInfoAllToMySQL;
import com.xiongmm.stream.sink.SinkInfoCmdToMySQL;
import com.xiongmm.stream.support.FlinkConstant;
import com.xiongmm.stream.support.FlinkSupportor;
import com.xiongmm.stream.support.OpentsdbConstant;
import com.xiongmm.stream.utils.PropertiesUtil;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;

import static com.xiongmm.stream.support.FlinkConstant.INT_1;
import static com.xiongmm.stream.support.FlinkConstant.INT_2;

/**
 * @author: XiongMM
 * @date: 2019/6/21 14:30
 * @description:
 */
public class InstanceInfoAllFlink {

    private final static Logger LOGGER = LoggerFactory.getLogger(InstanceInfoAllFlink.class);

    public static void main(String[] args) throws Exception {
        //这是一个入口类，可以用来设置参数和创建数据源以及提交任务
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        //告诉系统，用事件时间进行处理
        environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        //设置全局的并发数为1，后面在单独设置每个算子的并发度
        environment.setParallelism(INT_1);
        //设置数据源输入
        Properties props = new Properties();
        props.put(FlexQConstant.FLEXQ_ZKHOST, PropertiesUtil.getValue(FlinkConstant.ZK_AGENT));
        props.put(FlexQConstant.FLEXQ_ZKPREFIX, FlinkConstant.PREFIX);
        props.put(FlexQConstant.FLEXQ_TOPIC, FlinkConstant.INFO_ALL_TOPIC);
        props.put(FlexQConstant.FLEXQ_GROUPID, FlinkConstant.TOPIC_GROUP);
        props.put(FlexQConstant.FLEXQ_FORMAT, FlexQConstant.FORMAT_STRING);
        FlexQSource flexQSource = new FlexQSource(props);

        //1、得到计算好的流
        SingleOutputStreamOperator<InstanceInfoAllVo> dataStream
                //设置mq为数据源
                = environment.addSource(flexQSource)
                //String类型的info转化为对象
                .map((MapFunction<String, InstanceInfoAllVo>) info -> {
                    InstanceInfoAllVo infoAllVo = null;
                    try {
                        infoAllVo = JSON.parseObject(info, InstanceInfoAllVo.class);
                    } catch (Exception ex) {
                        LOGGER.error("analysis info to object has an error ", ex);
                    }
                    return infoAllVo;
                })
                //过滤掉为null的
                .filter((FilterFunction<InstanceInfoAllVo>) Objects::nonNull)
                //设置允许的最大延迟70秒，70秒一个Watermark (我的本意是第三分钟数据来的时候，形成新的Watermark)
                //当前最大事件时间 与 上一个Watermark的时间 的差值，如果超过了70秒，就会触发生成新的Watermark,相应的window随着watermark生成而被触发
                .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<InstanceInfoAllVo>(Time.seconds(70)) {
                    @Override
                    public long extractTimestamp(InstanceInfoAllVo element) {
                        //获取event time ( unix timestamp -> millis time)
                        return element.getCollectTime() * 1000;
                    }
                });

        //2、得到info stream
        SingleOutputStreamOperator<List<InstanceInfoAllVo>> infoStream
                //设置keyby,这里按ip:port进行分组，转化为KeyedStream
                = dataStream.keyBy((KeySelector<InstanceInfoAllVo, String>) value -> value.getIp() + ":" + value.getPort())
                //每当来一条info all信息，计算最近2条数据的差值
                .countWindow(INT_2, INT_1)
                //两条进行差值计算
                .reduce((ReduceFunction<InstanceInfoAllVo>) (infoAllVo1, infoAllVo2) -> {
                    List<InstanceInfoAllVo> list = new ArrayList<>();
                    list.add(infoAllVo1);
                    list.add(infoAllVo2);
                    list = FlinkSupportor.listOrder(list, INT_2);
                    return FlinkSupportor.infoAllVoInternalCalculate(list.get(0), list.get(2));
                })
                //聚合一分钟内的所有数据
                .timeWindowAll(Time.minutes(1))
                //发送一分钟内的所有数据
                .apply((AllWindowFunction<InstanceInfoAllVo, List<InstanceInfoAllVo>, TimeWindow>) (window, values, out) -> {
                    List<InstanceInfoAllVo> infoAllVos = new ArrayList<>();
                    for (InstanceInfoAllVo infoAllVo : values) {
                        if (infoAllVo == null) {
                            continue;
                        }
                        infoAllVos.add(infoAllVo);
                    }
                    if (infoAllVos.size() == 0) {
                        return;
                    }
                    out.collect(infoAllVos);
                });

        //3、得到command stream
        SingleOutputStreamOperator<List<InstanceInfoCmdVo>> commandStream
                //设置keyBy,这里按ip:port进行分组，转化为KeyedStream
                = dataStream.keyBy((KeySelector<InstanceInfoAllVo, String>) value -> value.getIp() + ":" + value.getPort())
                //每当来一条info all信息，计算最近2条数据的差值
                .countWindow(INT_2, INT_1)
                .apply((WindowFunction<InstanceInfoAllVo, List<InstanceInfoCmdVo>, String, GlobalWindow>) (key, window, input, out) -> {
                    List<InstanceInfoAllVo> list = new ArrayList<>();
                    for (InstanceInfoAllVo infoAllVo : input) {
                        list.add(infoAllVo);
                    }
                    if (list.size() < INT_2) {
                        LOGGER.error("window data not equals 2");
                        return;
                    }
                    list = FlinkSupportor.listOrder(list, INT_2);
                    List<InstanceInfoCmdVo> cmdVos = FlinkSupportor.infoAllCommandStatsInternalCalculate(list.get(0), list.get(1));
                    out.collect(cmdVos);
                });


        //info all信息入库mysql/opentsdb
        infoStream.addSink(new SinkInfoAllToMySQL());
        infoStream.map((MapFunction<List<InstanceInfoAllVo>, Collection<Point>>) FlinkSupportor::analyzeInfoToPoint).addSink(new OpentsdbSink(OpentsdbConstant.PROJECT_NAME, OpentsdbConstant.OPEN_TSDB_CLUSTER));

        //command 信息入库mysql/opentsdb
        commandStream.addSink(new SinkInfoCmdToMySQL());
        commandStream.map((MapFunction<List<InstanceInfoCmdVo>, Collection<Point>>) FlinkSupportor::processCommandStatsToOpenTsdb).addSink(new OpentsdbSink(OpentsdbConstant.PROJECT_NAME, OpentsdbConstant.OPEN_TSDB_CLUSTER));

        //commit
        environment.execute("start ozono info all flink job ～～～");
    }
}
